package com.example.components.utils;

import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by lkj on 2018/4/10.
 */
public final class InfluxdbWriter {

    private static final Logger logger = LoggerFactory.getLogger(InfluxdbWriter.class);

    /**
     * 缓存中的待入库的point数量上限，超过该数值就入一次库
     */
    private static final int MAX_BATCH_POINT_SIZE = 100;

    /**
     * 入库间隔时间，超过该间隔就入一次库
     */
    private static final int MAX_BATCH_POINT_INTERVAL_TIME = 10 * 1000;

    // =========================== 原有列表 ============================
    private static final List<Point> kpiPointList = Collections.synchronizedList(new ArrayList<>());

    private static final List<Point> eventPointList = Collections.synchronizedList(new ArrayList<>());
    // ================================================================

    // =========================== 新增列表 =============================

    // 带库名的结构
    private static final ConcurrentHashMap<String, List<Point>> kpiPointMap = new ConcurrentHashMap<>();

    private static final ConcurrentHashMap<String, List<Point>> agentStatusMap = new ConcurrentHashMap<String, List<Point>>();

    private static long lastWriteKpiTime = System.currentTimeMillis();

    // =================================================================
    private static long lastWriteEventTime = System.currentTimeMillis();

    private static long lastWriteKpiMapTime = System.currentTimeMillis();

    private static long lastAgentStatusMapTime = System.currentTimeMillis();

    private InfluxdbWriter() {
    }

    public static ConcurrentHashMap<String, List<Point>> getAgentStatusMap() {
        return agentStatusMap;
    }

    public static void write() {
        // 批量写kpi point,数量>10或者间隔>10s
        if (!kpiPointList.isEmpty() && (kpiPointList.size() > MAX_BATCH_POINT_SIZE
                || System.currentTimeMillis() - lastWriteKpiTime > MAX_BATCH_POINT_INTERVAL_TIME)) {
            lastWriteKpiTime = System.currentTimeMillis();
            // BatchPoints kpiPoints = BatchPoints.database(ZMQConst.INFLUX_DB_KPI)
            // .retentionPolicy(ZMQConst.INFLUX_RETENTION_KPI).points(getKpiPointList()).build();
            // writePoints(kpiPoints);
        }
        // 批量写event point
        if (!eventPointList.isEmpty() && (eventPointList.size() > MAX_BATCH_POINT_SIZE
                || System.currentTimeMillis() - lastWriteEventTime > MAX_BATCH_POINT_INTERVAL_TIME)) {
            lastWriteEventTime = System.currentTimeMillis();
            // BatchPoints eventPoints = BatchPoints.database(ZMQConst.INFLUX_DB_EVENTS)
            // .retentionPolicy(ZMQConst.INFLUX_RETENTION_EVENTS).points(getEventPointList()).build();
            // writePoints(eventPoints);
        }

        // 批量写新格式的记录
        if (pointMapoverSize() || System.currentTimeMillis() - lastWriteKpiMapTime > MAX_BATCH_POINT_INTERVAL_TIME) {
            lastWriteKpiMapTime = System.currentTimeMillis();
            for (String dbName : kpiPointMap.keySet()) {

                List<Point> pointList = kpiPointMap.get(dbName);
                if (!pointList.isEmpty()) {

                    // BatchPoints newFormatKpiPoints = BatchPoints.database(dbName)
                    // .retentionPolicy(ZMQConst.INFLUX_RETENTION_KPI).points(getArraysFromPointList(pointList))
                    // .build();
                    // writePoints(newFormatKpiPoints);

                }
            }
        }

        // 写agent状态
        long agentWriteInterval = 3 * 60 * 1000L; // 3分钟写一次
        if (System.currentTimeMillis() - lastAgentStatusMapTime > agentWriteInterval && agentStatusMap.size() > 0) {
            lastAgentStatusMapTime = System.currentTimeMillis();
            writeAgentStatusBatch();
        }

    }

    public static void writeAgentStatusBatch() {
        for (String dbName : agentStatusMap.keySet()) {
            List<Point> pointList = agentStatusMap.get(dbName);
            if (!pointList.isEmpty()) {

                // BatchPoints newFormatKpiPoints = BatchPoints.database(dbName)
                // .retentionPolicy(ZMQConst.INFLUX_RETENTION_KPI).points(getArraysFromPointList(pointList)).build();
                // writePoints(newFormatKpiPoints);
            }
        }
    }

    /**
     * map中的总数超限
     *
     * @return
     */
    private static boolean pointMapoverSize() {

        int size = 0;
        for (String dbName : kpiPointMap.keySet()) {
            List<Point> pointList = kpiPointMap.get(dbName);
            size += pointList.size();
            if (size > MAX_BATCH_POINT_SIZE) {
                return true;
            }
        }

        return false;

    }

    /**
     * writePoints
     */
    private static void writePoints(BatchPoints batchPoints) {
        try {
            InfluxDBUtil.batchWrite(batchPoints);
        } catch (RuntimeException e) {
            logger.error(e.getMessage());
        }
    }

    /**
     * addKpiPointList
     *
     * @param point Point
     */
    static void addKpiPointList(Point point) {
        kpiPointList.add(point);
    }

    /**
     * addEventPointList
     *
     * @param point Point
     */
    static void addEventPointList(Point point) {
        eventPointList.add(point);
    }

    /**
     * 返回深copy，并清除
     *
     * @return Point[]
     */
    private static Point[] getKpiPointList() {
        List<Point> listCopy = new ArrayList<>();
        synchronized (kpiPointList) {
            listCopy.addAll(kpiPointList);
            kpiPointList.clear();
        }
        return listCopy.toArray(new Point[listCopy.size()]);
    }

    /**
     * 返回深copy，并清除
     *
     * @return Point[]
     */
    private static Point[] getEventPointList() {
        List<Point> listCopy = new ArrayList<>();
        synchronized (eventPointList) {
            listCopy.addAll(eventPointList);
            eventPointList.clear();
        }
        return listCopy.toArray(new Point[listCopy.size()]);
    }

    /**
     * 返回深copy，并清除
     *
     * @return Point[]
     */
    private static Point[] getArraysFromPointList(List<Point> pointList) {
        List<Point> listCopy = new ArrayList<>();
        synchronized (pointList) {
            listCopy.addAll(pointList);
            pointList.clear();
        }
        return listCopy.toArray(new Point[listCopy.size()]);
    }

    /**
     * 增加单个point
     *
     * @param dbName 数据库名
     * @param point  Point
     */
    public static void addOnePoint(String dbName, Point point) {

        boolean existFlag = false;
        /*
         * for (String name : kpiPointMap.keySet()) { if (dbName.equals(name)) { kpiPointMap.get(name).add(point);
         * existFlag = true; break; } }
         */
        for (Iterator<Map.Entry<String, List<Point>>> itertor = kpiPointMap.entrySet().iterator(); itertor.hasNext(); ) {
            Map.Entry<String, List<Point>> entry = itertor.next();
            if (dbName.equals(entry.getKey())) {
                kpiPointMap.get(entry.getKey()).add(point);
                existFlag = true;
                break;
            }
        }
        if (!existFlag) {
            List<Point> pointList = Collections.synchronizedList(new ArrayList<>());
            pointList.add(point);
            kpiPointMap.put(dbName, pointList);
        }
    }

    public static void addOnePoint(Map<String, List<Point>> map, String dbName, Point point) {
        boolean existFlag = false;
        /*
         * for (String name : map.keySet()) { if (dbName.equals(name)) { map.get(name).add(point); existFlag = true;
         * break; } }
         */
        for (Iterator<Map.Entry<String, List<Point>>> iterator = map.entrySet().iterator(); iterator.hasNext(); ) {
            Map.Entry<String, List<Point>> entry = iterator.next();
            if (dbName.equals(entry.getKey())) {
                map.get(entry.getKey()).add(point);
                existFlag = true;
                break;
            }
        }
        if (!existFlag) {
            List<Point> pointList = Collections.synchronizedList(new ArrayList<>());
            pointList.add(point);
            map.put(dbName, pointList);
        }
    }

    /**
     * 增加point集合，必须是同一个数据库
     *
     * @param dbName    数据库名
     * @param pointList pointList
     */
    public static void addPointList(String dbName, List<Point> pointList) {

        boolean existFlag = false;
        for (String name : kpiPointMap.keySet()) {
            if (dbName.equals(name)) {
                kpiPointMap.get(name).addAll(pointList);
                existFlag = true;
                break;
            }
        }
        if (!existFlag) {
            List<Point> newPointList = Collections.synchronizedList(new ArrayList<>());
            newPointList.addAll(pointList);
            kpiPointMap.put(dbName, newPointList);
        }

    }

}
